﻿using LJ.EventBus.FastDog.Data;
using LJ.EventBus.FastDog.Transport.Abstractions;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace LJ.EventBus.FastDog.Transport.InMemoryQueue.Internal;


public class InMemoryQueueConsumerClient : IConsumerClient
{
    private readonly MemoryQueue _memoryQueue;
    private readonly string _groupId;
    private readonly BlockingCollection<TransportMessage> _messageQueue = [];

    public InMemoryQueueConsumerClient(MemoryQueue memoryQueue, string groupId)
    {
        _memoryQueue = memoryQueue;
        _groupId = groupId;
        _memoryQueue.RegisterConsumerClient(groupId, this);
    }

    public Func<TransportMessage, object, Task>? OnMessageCallback { get; set; }
    public Action<LogMessageEventArgs>? OnLogCallback { get; set; }

    public ICollection<string> FetchTopics(IEnumerable<string> topics)
    {
        _memoryQueue.Subscribe(_groupId, topics);
        return topics.ToList();
    }

    public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
    {
        foreach (var message in _messageQueue.GetConsumingEnumerable(cancellationToken))
        {
            OnMessageCallback?.Invoke(message, _messageQueue).ConfigureAwait(false).GetAwaiter().GetResult();
        }
    }

    public void Commit(object? sender)
    {
    }

    public void Reject(object? sender)
    {
    }

    public void Subscribe(IEnumerable<string> topics)
    {
        //_memoryQueue.Subscribe(_groupId, topics);
    }
    public void AddSubscribeMessage(TransportMessage message)
    {
        _messageQueue.Add(message);
    }

    public void Dispose()
    {
        _memoryQueue.Unsubscribe(_groupId);
        GC.SuppressFinalize(this);
    }
}
